package com.niodata.dp.task.log;

import com.google.common.base.Strings;
import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;
import com.niodata.dp.core.ConfigurationKeys;
import com.niodata.dp.core.DpConfiguration;
import com.niodata.dp.core.exception.InvalidParamException;
import com.niodata.dp.core.exception.ResourceNotFoundException;
import com.niodata.dp.log.LogUtil;
import com.niodata.dp.log.viewer.Host;
import com.niodata.dp.log.viewer.JschLogViewer;
import com.niodata.dp.log.viewer.ViewerSession;
import com.niodata.dp.task.dao.JobLogMappingMapper;
import com.niodata.dp.task.dao.JobMapper;
import com.niodata.dp.task.entity.JobInfo;
import com.niodata.dp.task.entity.JobLogMapping;
import com.niodata.dp.util.StringUtil;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component("taskLogManager")
public class TaskLogManager {

  private static final Logger logger = Logger.getLogger(TaskLogManager.class);
  @Autowired
  JobMapper jobMapper;
  @Autowired
  JobLogMappingMapper jobLogMapper;
  private JschLogViewer logViewer = new JschLogViewer();
  @Value("${" + ConfigurationKeys.SSH_USER + ":root}")
  private String sshUser;
  private String sshPassword =
      DpConfiguration.getString(ConfigurationKeys.SSH_PASSWORD, null);
  private String sshIdRsaPath =
      DpConfiguration.getString(ConfigurationKeys.SSH_IDRSA_PATH, null);
  @Value("${" + ConfigurationKeys.SSH_PORT + ":22}")
  private String sshPort;

  /**
   * connect to log file,return log viewer session.
   *
   * @param taskId task id
   * @param execId exec id
   * @param jobName jobname
   * @param viewMode viewmode
   * @param connectParam connect param
   * @param logType log file type
   * @return session
   * @throws IOException exp
   */
  public ViewerSession connectLog(String taskId, int execId, String jobName, LogViewMode viewMode,
      String connectParam, LogType logType) throws IOException {
    JobLogMapping jobLogInfo = null;
    if (LogType.TASKLOG.equals(logType)) {
      jobLogInfo = jobLogMapper
          .getJobLogMapping(String.valueOf(execId), null, logType);
    } else if (LogType.YARNLOG_ERR.equals(logType) || LogType.YARNLOG_OUT.equals(logType)) {
      jobLogInfo = jobLogMapper
          .getJobLogMapping(String.valueOf(execId), jobName, LogType.YARNLOG);
    }

    if (null == jobLogInfo) {
      throw new ResourceNotFoundException("暂时没有此任务的日志信息，请稍等");
    }
    Host host = new Host();
    host.setIp(jobLogInfo.getHostAddr());
    host.setPort(Integer.valueOf(sshPort));
    host.setUser(sshUser);
    host.setPassword(sshPassword);
    host.setIdrsaPath(sshIdRsaPath);
    String logfile = "";
    if (LogType.TASKLOG.equals(logType)) {
      String baseDir = jobLogInfo.getLogDir();
      String log = getMaxLogFile(host.getIp(), baseDir, jobName);
      if (StringUtil.isEmpty(log)) {
        throw new ResourceNotFoundException("没有日志文件");
      }
      logfile = baseDir + File.separator + log;
    } else if (LogType.YARNLOG_ERR.equals(logType)) {
      logfile = jobLogInfo.getLogDir() + File.separator + "stderr";
    } else if (LogType.YARNLOG_OUT.equals(logType)) {
      logfile = jobLogInfo.getLogDir() + File.separator + "stdout";
    }
    logger.info("execid:" + execId + "," + jobName + ",logfile:" + logfile);
    switch (viewMode) {
      case TAILLINES:
        return logViewer.tailFileLines(host, logfile, Integer.parseInt(connectParam));
      case HEADLINES:
        return logViewer.headFileLines(host, logfile, Integer.parseInt(connectParam));
      case TAILF:
        return logViewer.tailFile(host, logfile);
      case GREP:
        return logViewer.grepFile(host, logfile, connectParam);
      default:
    }
    return null;
  }

  public void checkJobInfo(String taskId, String jobName) {
    JobInfo jobInfo = jobMapper.getJobByNameByTaskId(jobName, taskId);
    if (null == jobInfo) {
      throw new InvalidParamException("param error!");
    }
  }

  private String getMaxLogFile(String host, String dir, String jobName) {
    List<String> fileList = new ArrayList();
    JSch jsch = new JSch();
    Session session = null;
    ChannelExec channel = null;
    String cmd = "ls -1 " + dir + "| grep " + jobName + ".log";
    try {
      if (!Strings.isNullOrEmpty(sshIdRsaPath)) {
        jsch.addIdentity(sshIdRsaPath);
      }
      session = jsch.getSession(sshUser, host, Integer.valueOf(sshPort));
      session.setConfig("PreferredAuthentications", "publickey,password");
      if (!Strings.isNullOrEmpty(sshPassword)) {
        session.setPassword(sshPassword);
      }
      session.setConfig("StrictHostKeyChecking", "no");
      session.connect();
      channel = (ChannelExec) session.openChannel("exec");
      channel.setPty(true);
      InputStream in = channel.getInputStream();
      channel.setCommand(cmd);
      channel.connect();

      BufferedReader reader = null;
      try {
        reader = new BufferedReader(new InputStreamReader(in));
        String line = null;
        while ((line = reader.readLine()) != null) {
          if (line.startsWith("_job") && line.endsWith(".log")) {
            fileList.add(line);
          }
        }
      } catch (Exception e) {
        LogUtil.logStackTrace(logger, e);
      }

    } catch (Exception e) {
      logger.error("getMaxLogFile error:" + host + "," + cmd);
      LogUtil.logStackTrace(logger, e);
    } finally {
      channel.disconnect();
      session.disconnect();
    }

    Collections.sort(fileList, new Comparator<String>() {
      @Override
      public int compare(String o1, String o2) {
        if (o1.length() > o2.length()) {
          return -1;
        } else if (o1.length() < o2.length()) {
          return 1;
        } else {
          if (o1.compareTo(o2) > 0) {
            return -1;
          } else if (o1.compareTo(o2) < 0) {
            return 1;
          }
          return 0;
        }

      }
    });
    if (null == fileList || fileList.size() < 1) {
      return null;
    }
    return fileList.get(0);
  }

}
